package com.eshipenet.shipnet.edgeboxextbuilder
import org.apache.flink.streaming.api.datastream.CoGroupedStreams
import org.apache.flink.streaming.api.windowing.triggers.{Trigger, TriggerResult}
import org.apache.flink.streaming.api.windowing.windows.{GlobalWindow, Window}

object EdgeBoxModelChangeTrigger {

  type T1 = (String, Set[EdgeBox])
  type T2 = EdgeBoxModelAction

  type U = CoGroupedStreams.TaggedUnion[T1, T2]
}

import EdgeBoxModelChangeTrigger._

class EdgeBoxModelChangeTrigger extends Trigger[U, GlobalWindow] {
  override def onElement(element: U,
                         timestamp: Long,
                         window: GlobalWindow,
                         ctx: Trigger.TriggerContext): TriggerResult = {
    if (element.isTwo) {
      TriggerResult.FIRE_AND_PURGE
    } else {
      TriggerResult.CONTINUE
    }
  }
  override def onProcessingTime(time: Long, window: GlobalWindow, ctx: Trigger.TriggerContext): TriggerResult =
    TriggerResult.CONTINUE

  override def onEventTime(time: Long, window: GlobalWindow, ctx: Trigger.TriggerContext): TriggerResult =
    TriggerResult.CONTINUE

  override def clear(window: GlobalWindow, ctx: Trigger.TriggerContext): Unit = {}
}
